package com.youshang520i.learningmysqlcanal.config;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import com.youshang520i.learningmysqlcanal.model.Goods;
import com.youshang520i.learningmysqlcanal.model.OperationRecord;
import com.youshang520i.learningmysqlcanal.service.OperationRecordService;
import org.apache.zookeeper.Op;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;
import java.sql.JDBCType;
import java.time.LocalDate;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * canal 监听客户端
 * @author youshang
 */
@Component
public class CanalClient {

    @Autowired
    private OperationRecordService operationRecordService;



    private static String SERVER_ADDRESS = "192.168.3.123";
    private static Integer PORT = 11111;
    private static String DESTINATION = "example";
    private static String USERNAME = "root";
    private static String PASSWORD = "123456";

    @PostConstruct
    public  void canalInit() {
        // newSingleConnector是简单的ip直连模式，基于CanalServerWithNetty定义的网络协议接口，对于canal数据进行get/rollback/ack等操作
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(SERVER_ADDRESS, PORT), DESTINATION, USERNAME, PASSWORD);
        canalConnector.connect();
        //订阅 全部数据库全部表 -> ".*\\..*"
        //数据库.表
        canalConnector.subscribe("db_youshang.*");
        //恢复到之前同步的数据，避免误操作
        canalConnector.rollback();

        while (true) {
            // 获取指定数量的数据，但是不做确认
            Message message = canalConnector.getWithoutAck(100);
            // 消息ID
            long batchId = message.getId();
            if (batchId != -1) {
                // System.out.println(message.getEntries());
                System.out.println("msgId-->" + batchId);
                printEntity(message.getEntries());
//                 canalConnector.ack(batchId); //提交确认
//                 canalConnector.rollback(); //处理失败，回滚数据
            }
        }
    }
    public  void printEntity(List<CanalEntry.Entry> entries){
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA){
                continue;
            }
            try{
                // 数据反序列化
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());

                String tableName = entry.getHeader().getTableName();
                for (CanalEntry.RowData rowData: rowChange.getRowDatasList()) {
                    switch (rowChange.getEventType()){
                        case INSERT:
                            //测试users表进行映射处理
                            Goods userDTO = CanalDataHadler.convertToBean(rowData.getAfterColumnsList(), Goods.class);
                            System.out.println("执行了一条插入操作，数据为：" + userDTO);
                            break;
                        case DELETE:
                            Goods deleteUserDTO = CanalDataHadler.convertToBean(rowData.getAfterColumnsList(), Goods.class);
                            System.out.println("执行了一条删除操作，数据为：" + deleteUserDTO);
                            break;
                        case UPDATE:
                            operationRecordService.insertSelective(rowChange.getRowDatasList(),tableName);
                            break;
                        default:
                            break;
                    }
                }
            }catch (InvalidProtocolBufferException e){
                e.printStackTrace();
            }
        }
    }

}